package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

object Demo21Cache {
  def main(args: Array[String]): Unit = {

    /**
     * Cache：缓存，为了加快执行速度提高效率
     */

    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName(this.getClass.getSimpleName.replace("$", ""))
    val sc = new SparkContext(conf)

    val stuRDD: RDD[String] = sc.textFile("scala/data/stu/students.txt")

    val mapStuRDD: RDD[String] = stuRDD.map(line => {
      println("执行了map方法")
      line
    })

    // 对多次使用的RDD可以进行缓存，默认会使用内存进行缓存
    //    mapStuRDD.cache() // 默认使用MEMORY_ONLY
    // 内存不够怎么办？选择合适的缓存策略，策略因素：是否用磁盘、是否用内存、是否用堆外内存、是否序列化、是否需要副本
    /**
     * 尽量将更多的数据缓存在内存
     * 内存够不够？
     * 足够 -> MEMORY_ONLY
     * 不够 -> MEMORY_AND_DISK_SER
     */
    mapStuRDD.persist(StorageLevel.MEMORY_AND_DISK_SER)


    // 统计班级人数
    mapStuRDD.map(line => (line.split(",")(4), 1)).reduceByKey(_ + _).foreach(println)

    // 统计性别人数
    mapStuRDD.map(line => (line.split(",")(3), 1)).reduceByKey(_ + _).foreach(println)

    // 统计年龄人数
    mapStuRDD.map(line => (line.split(",")(2), 1)).reduceByKey(_ + _).foreach(println)

    // 实时计算任务，程序是无法停止的，一定要记得释放缓存
    mapStuRDD.unpersist()

    while (true) {

    }

  }

}
